package com.shujia.core

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.io.Source

/**
 *  广播大【数据量大】变量
 */
object BroadcastDemo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("广播变量")

    val sc = new SparkContext(conf)

    //scala的方式读取学生数据，Driver端
    val lineList: List[String] = Source.fromFile("spark/data/students.txt").getLines().toList
    val studentKVMap: Map[String, String] = lineList.map((s: String) => {
      val info: Array[String] = s.split(",")
      (info(0), info.mkString(","))
    }).toMap

    /**
     *  将studentKVMap广播到所有的Executor中，将来任务执行的时候，直接从本节点中的Executor取数据即可
     *  不需要每次都从Driver端拉取数据了。
     */
    val bc: Broadcast[Map[String, String]] = sc.broadcast(studentKVMap)

    val scoRDD: RDD[String] = sc.textFile("spark/data/score.txt")
    val resRDD: RDD[(String, String, String, String)] = scoRDD.map((s: String) => {
      val infos: Array[String] = s.split(",")
      val id: String = infos(0)
      //从map集合，通过键获取学生的信息
      val stuInfo: String = bc.value.getOrElse(id, "无")
      (id, stuInfo, infos(1), infos(2))
    })

    resRDD.foreach(println)

  }
}
